package com.mall.seckill.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.OrderCoreService;
import com.mall.order.OrderPromoService;
import com.mall.order.constant.OrderRetCode;
import com.mall.order.dto.CreateOrderRequest;
import com.mall.order.dto.CreateOrderResponse;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.order.dto.CreateSeckillOrderResponse;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;

/**
 * @author lichao
 * @createdate 2021-02-01 21:46
 */
@Component
public class MqMessageConsumer {


    private DefaultMQPushConsumer mqPushConsumer;
    @Reference(interfaceClass = OrderCoreService.class,check = false)
    OrderPromoService orderPromoService;

    @PostConstruct
    public void init(){

        mqPushConsumer = new DefaultMQPushConsumer("seckill_producer");
        mqPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        mqPushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String bodyStr = new String(body);
                HashMap map = JSON.parseObject(bodyStr, HashMap.class);
                CreateSeckillOrderRequest request = (CreateSeckillOrderRequest) map.get("request");
                CreateSeckillOrderResponse response = orderPromoService.createSeckillOrder(request);

                String code = response.getCode();
                if(code.equals(OrderRetCode.SUCCESS)){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
